package day19;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

public class RealTimeTest {
    public static void main(String[] args) throws Exception {
        /**
         *
         * 需求：流数据的WordCount。使用netcat输入命令：nc -l -p 6666。然后启动本程序，
         * 然后在netcat里输入数据，在本程序控制台里就会输出wc
         * */
        // 1.得到流式处理的环境
        StreamExecutionEnvironment senv = StreamExecutionEnvironment.getExecutionEnvironment();

        // 2.监听流数据的来源（使用socket）
        DataStream<String> dataStream = senv.socketTextStream("master",6666);

        SingleOutputStreamOperator<Tuple2<String, Integer>> pairs = dataStream.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public void flatMap(String s, Collector<Tuple2<String, Integer>> collector) throws Exception {
                for (String word : s.split(" ")) {
                    collector.collect(new Tuple2<String, Integer>(word, 1));
                }
            }
        });

        // 以元祖的第一个数分组
        KeyedStream<Tuple2<String, Integer>, Tuple> keyByed = pairs.keyBy(0);
        // 时间窗口：timeWindow(Time.seconds(5)),second(5)是5秒执行一次。可以按天days、时hours、分minutes、秒seconds、毫秒来执行。（注意这里仅仅是统计5秒内的wc，而不是全部累加窗口函数）
        WindowedStream<Tuple2<String, Integer>, Tuple, TimeWindow> windowOperations = keyByed.timeWindow(Time.seconds(5));
        // 对时间窗口如何操作：这里用的是sum(1)即对元祖第二个参数相加
        SingleOutputStreamOperator<Tuple2<String, Integer>> sumed = windowOperations.sum(1);

        sumed.print();
        senv.execute(); //注意要抛出错误。throws Exception

    }
}
